package com.huatai.bi.mqtt;

import com.alibaba.nacos.api.utils.NetUtils;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.huatai.bi.constant.RedisConstant;
import com.huatai.bi.entity.MqMqttClientEntity;
import com.huatai.bi.service.MqMqttClientService;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @program: itc-airport-cloud
 * @description: mqtt监听
 * @author: songxiude
 * @create: 2024-07-15 15:24
 **/
@Slf4j
@Component
public class MqttClientListener implements ApplicationListener<ContextRefreshedEvent> {


	@Autowired
	private MqMqttClientService mqttClientService;

	@Autowired
	private RedissonClient redissonClient;

	@Value("${server.port}")
	private String port;
	private final AtomicBoolean isInit = new AtomicBoolean(false);

	@Override
	public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
		//防止重复触发
		if (!isInit.compareAndSet(false, true)) {
			return;
		}
		String ipport = NetUtils.localIP() + ":" + port;
		LambdaQueryWrapper<MqMqttClientEntity> query = Wrappers.lambdaQuery();
		query.and(wrapper -> wrapper.eq(MqMqttClientEntity::getEnableFlag, true).eq(MqMqttClientEntity::getStatus, 0));
		query.or(wrapper -> wrapper.eq(MqMqttClientEntity::getIpPort, ipport));
		//初始化链接mqtt
//		List<MqMqttClientEntity> list = mqttClientService.list(Wrappers.<MqMqttClientEntity>lambdaQuery()
//			.eq(MqMqttClientEntity::getEnableFlag, true)
//			.eq(MqMqttClientEntity::getStatus, 0));

		List<MqMqttClientEntity> list = mqttClientService.list(query);
		if (!CollectionUtils.isEmpty(list)) {
			for (MqMqttClientEntity mqtt : list) {
				RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(RedisConstant.MQTT_CLIENT + mqtt.getId());
				RLock rLock = readWriteLock.writeLock();
				rLock.lock();
				try {
					//再次确认是否被其他服务连接
					MqMqttClientEntity mqttClient = mqttClientService.getById(mqtt.getId());
					if ((mqttClient.getStatus().intValue() == 0 && mqttClient.getEnableFlag()) || mqttClient.getIpPort().equals(ipport)) {
						boolean b = mqttClientService.initMqttServer(mqttClient);
						if (b) {
							mqttClient.setStatus(1);
							mqttClient.setIpPort(ipport);
							mqttClientService.updateById(mqttClient);
						}
					}
				} catch (Exception e) {
					log.error(e.getMessage(), e);
				} finally {
					rLock.unlock();
				}
			}
		}

		log.info("该实例创建了" + MqttClientConnect.mqttClients.size() + "mqtt连接");
	}

}
